package cn.terminus.service.customer.common;

import cn.terminus.result.common.serial.UUIDSerial;
import cn.terminus.result.common.util.TerminusDateUtils;
import cn.terminus.service.customer.analysis.KafkaTopicProperties;
import cn.terminus.service.customer.analysis.OverviewAnalysis;
import cn.terminus.service.customer.kafka.KafkaContractParam;
import cn.terminus.service.customer.kafka.KafkaCustomerChannelParam;
import cn.terminus.service.customer.kafka.KafkaCustomerParam;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.MessageSource;
import org.springframework.context.i18n.LocaleContextHolder;
import org.springframework.kafka.core.KafkaTemplate;

import javax.annotation.Resource;

/**
 * Service基类
 *
 * @author ty
 **/
public class AbstractTerminusService {

    /**
     * 日志
     */
    protected final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * access token
     */
    protected static final String ACCESS_TOKEN_KEY = "access_token:%s-%s";

    /**
     * 消息源资源对象
     */
    @Autowired
    private MessageSource messageSource;

    /**
     * 通用的唯一编号生成器
     */
    @Autowired(required = false)
    private UUIDSerial uuidSerial;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    private KafkaTopicProperties topicProperties;

    /**
     * 获取对应编码的信息描述
     *
     * @param code 编码
     * @return
     */
    public String getMessage(String code) {
        return this.messageSource.getMessage(code, null, "", LocaleContextHolder.getLocale());
    }

    /**
     * 生成新的uuid
     *
     * @return
     */
    public Long genUUID() {
        if (null != uuidSerial) {
            return Long.valueOf(uuidSerial.uuid());
        }
        return Long.valueOf(TerminusDateUtils.getUUID());
    }

    /**
     * 概览消息推送
     * @param param
     */
    public void sendOverviewKafka(OverviewAnalysis param) {
        kafkaTemplate.send(topicProperties.getOverviewAnalysis(), topicProperties.getOverviewAnalysis(), JSON.toJSONString(param));
    }

    protected void sendCustomerKafka(KafkaCustomerParam param) {
        kafkaTemplate.send(topicProperties.getCustomerAnalysis(), topicProperties.getCustomerAnalysis(), JSON.toJSONString(param));
    }

    protected void sendContractKafka(KafkaContractParam param) {
        kafkaTemplate.send(topicProperties.getContractTrend(), topicProperties.getContractTrend(), JSON.toJSONString(param));
    }

    protected void sendChannelKafka(KafkaCustomerChannelParam param) {
        kafkaTemplate.send(topicProperties.getChannelAnalysis(), topicProperties.getChannelAnalysis(), JSON.toJSONString(param));
    }


}
